package edu.kit.ksri.db.mturk.simulations;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.FilenameFilter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.log4j.Logger;

import com.amazonaws.mturk.requester.Assignment;

import edu.kit.ksri.db.mturk.utils.messaging.Result;
import edu.kit.ksri.db.mturk.utils.messaging.ResultOutputCollector;
import edu.kit.ksri.db.mturk.var.Icd;
import edu.kit.ksri.db.mturk.var.SimMode;
import edu.kit.ksri.db.mturk.var.StaticInformation;
import edu.kit.ksri.db.mturk.workerpools.PoolManager;
import edu.kit.ksri.db.mturk.workerpools.SimWorker;
import edu.kit.ksri.db.mturk.workerpools.WorkerPool;

/**
 * Copyright (c) 2010-2011 Karlsruhe Institute of Technology (KIT), Germany
 * 
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 * 
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 * 
 * --------------------------------------------------------------------------
 * 
 * Software: CSP/WMV tool for dynamic quality management of microtasks
 * http://www.ksri.kit.edu/Upload/Users/PeopleClouds/
 * HCOMP2011_An_extendable_toolkit_for_managing_quality_of_human
 * -based_eServices.pdf
 * 
 * Organization: Karlsruhe Institute of Technology (KIT), Germany
 * http://www.kit.edu
 * 
 * Project: PeopleClouds project of the Karlsruhe Service Research
 * Institute(KSRI) http://www.ksri.kit.edu/Default.aspx?PageId=818
 * 
 * Contributing authors: David Bermbach, Sandra Rath, Pascal Wichmann, Robert
 * Kern
 */
public class SimMTurk implements Runnable {

	private static final Logger basiclog = Logger.getLogger("basic."
			+ SimMTurk.class);

	/** all hit ids */
	private static String[] hitIds;

	/** all results which can be retrieved */
	private static final ConcurrentHashMap<String, Queue<Assignment>> outputAnswers = new ConcurrentHashMap<String, Queue<Assignment>>();

	/**
	 * all possible HIT ids, the largest question id number must be equal to
	 * {@link PerfSimWorker}.possibleAnswers.length -1
	 */
	private static final Queue<String> queuedHitIds = new ConcurrentLinkedQueue<String>();

	/** contains all qualification values for workers */
	private static final Map<String, Integer> qualificationValues = new ConcurrentHashMap<String, Integer>();

	/** determines the end of the simulation */
	private static boolean running = true;

	/** references to all {@link PerfSimWorker} {@link Thread}s */
	private static final HashMap<String, Thread> workerThreads = new HashMap<String, Thread>();

	/** counts the number of added answers */
	private static int counter = 0;

	/** announced HITs */
	private static Queue<String> publishedHITs = new ConcurrentLinkedQueue<String>();

	/** not announced HITs */
	private static Queue<String> notPublishedHITs = new ConcurrentLinkedQueue<String>();

	/** prioritized HITs */
	private static Queue<Result> prioritizedHITs = new ConcurrentLinkedQueue<Result>();

	private static boolean isDoneWithInitializing = false;

	/**
	 * initializes and starts the {@link PerfSimMTurk}
	 * 
	 * @param file
	 *            filename where the input data for the workers can be found.
	 *            must be a comma-separated text file of the form worker_id
	 *            qualification_value + answers (all comma-separated) every
	 *            worker should have the same number of possible answers. This
	 *            method makes no safety checks and will crash if any condition
	 *            is violated. # marks a comment line.
	 * 
	 */
	public static void initialize(String inputDirectory) throws Exception {

		// Read all the files from the input directory
		File dir = new File(inputDirectory);
		if (dir.isDirectory()) {
			File[] allXmlFiles = dir.listFiles(new FilenameFilter() {
				public boolean accept(File dir, String name) {
					// Files have to start with "all.", otherwise the files
					// should not be included
					if (name.trim().startsWith("all."))
						return true;
					return false;
				}
			});

			// Initialize all the possible HIT IDs
			hitIds = new String[allXmlFiles.length];

			// Do for every file
			for (File f : allXmlFiles) {
				String hitid = "";
				try {
					// Read all lines in the XML file
					BufferedReader br = new BufferedReader(new FileReader(f));
					String content = "", line = br.readLine();
					while (line != null) {
						content += line;
						line = br.readLine();
					}

					// Read HIT ID
					int start = content.indexOf("hitid=") + 7;
					content = content.substring(start);
					int end = content.indexOf("\"");
					hitid = content.substring(0, end);

					// Read Mapping number
					start = content.indexOf("mappingNumber=") + 15;
					content = content.substring(start);
					end = content.indexOf("\"");
					int mappingNumber = Integer.parseInt(content.substring(0,
							end));

					// Add the HIT to the list of HITIDs for correct order
					hitIds[mappingNumber - 1] = hitid;

					// Read number of results
					start = content.indexOf("numberOfResults=") + 17;
					content = content.substring(start);
					end = content.indexOf("\"");
					int number = Integer.parseInt(content.substring(0, end));

					// Do that for every assignment in the XML file
					LinkedList<Assignment> assignments = new LinkedList<Assignment>();

					for (int i = 0; i < number; i++) {
						// Read worker ID
						start = content.indexOf("worker=") + 8;
						content = content.substring(start);
						end = content.indexOf("\"");
						String workerid = content.substring(0, end);

						// Read assignment ID
						start = content.indexOf("assignmentID=") + 14;
						content = content.substring(start);
						end = content.indexOf("\"");
						String assignmentID = content.substring(0, end);

						// Read submission time
						start = content.indexOf("submissionTime=") + 16;
						content = content.substring(start);
						end = content.indexOf("\"");
						content.substring(0, end);// submission time

						// Need to concatenate the (different than originally
						// read) XML answer for the assignment that has to be
						// constructed
						String xml = null;

						if (StaticInformation.scenario instanceof Icd) {

							// Concatenate the log entry
							String answersLog = "Reading siminput directory - Current HIT is "
									+ hitid
									+ " and assignment is "
									+ assignmentID + ". Answers are ";
							String answers = "";

							// Concatenate the XML answer
							xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?><QuestionFormAnswers xmlns=\"http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd\">";

							// Do that for every single answer in the assignment
							// (Consists of several labels * (several flags + 1
							// code) + 1 comment field
							for (int j = 0; j < Icd.maxNumberOfLabels
									* (Icd.flagCount + 1) + 1; j++) {
								// Read question ID
								start = content.indexOf("<QuestionIdentifier>") + 20;
								content = content.substring(start);
								end = content.indexOf("</QuestionIdentifier>");
								String questionId = content.substring(0, end);
								basiclog.debug(questionId);

								// Read the answer
								start = content.indexOf("<FreeText>") + 10;
								content = content.substring(start);
								end = content.indexOf("</FreeText>");
								String answer = content.substring(0, end);
								basiclog.debug(answer);

								// Concatenate the answers for log
								if (j == (Icd.maxNumberOfLabels * Icd.flagCount) - 1)
									answers = answers + answer;
								else
									answers = answers + answer + ", ";

								content = content.substring(end);

								// Concatenate the XML
								xml = xml + "<Answer><QuestionIdentifier>"
										+ questionId
										+ "</QuestionIdentifier><FreeText>"
										+ answer + "</FreeText></Answer>";
							}

							// Concatenate the XML
							xml = xml + "</QuestionFormAnswers>";

							// Log what we just read
							basiclog.info(answersLog + answers);
						} else {
							// OCR and Poker consist only of one Question
							// Identifier and one FreeText
							// Read question ID
							start = content.indexOf("<QuestionIdentifier>") + 20;
							content = content.substring(start);
							end = content.indexOf("</QuestionIdentifier>");
							String questionId = content.substring(0, end);

							// Read the answer
							start = content.indexOf("<FreeText>") + 10;
							content = content.substring(start);
							end = content.indexOf("</FreeText>");
							String answer = content.substring(0, end);

							content = content.substring(end);

							// Log what we just read
							basiclog.info("Reading siminput directory - Current HIT is "
									+ hitid
									+ " and assignment is "
									+ assignmentID + ". Answer is " + answer);

							// Concatenate the XML answer
							xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?><QuestionFormAnswers xmlns=\"http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd\"><Answer>"
									+ "<QuestionIdentifier>"
									+ questionId
									+ "</QuestionIdentifier><FreeText>"
									+ answer
									+ "</FreeText></Answer></QuestionFormAnswers>";
						}

						// Create new assignment
						Assignment assignment = new Assignment();
						assignment.setAnswer(xml);
						assignment.setWorkerId(workerid);
						assignment.setAssignmentId(assignmentID);
						assignment.setHITId(hitid);

						// Add the assignment to the assignment collection
						assignments.add(assignment);

						// Add this worker to the active workerpool
						WorkerPool activeWorkerPool = PoolManager.getInstance()
								.getPoolForName("ACTIVE");
						SimWorker currentWorker = new SimWorker(workerid);
						if (!activeWorkerPool.containsWorker(currentWorker)) {
							activeWorkerPool.addWorker(currentWorker);
							// workerThreads.add(new Thread(currentWorker));
							// addWorkerThread(currentWorker);
							// addWorkerThread(((SimWorker)
							// (activeWorkerPool.getWorkerForID(workerid))));
						}

						// Add assignment to this worker's assignment collection
						((SimWorker) (activeWorkerPool.getWorkerForID(workerid)))
								.addAssignment(hitid, assignment);

						boolean success = ((SimWorker) (activeWorkerPool
								.getWorkerForID(workerid)))
								.hasAssignments(hitid);

						
						SimWorker bla = ((SimWorker) (activeWorkerPool.getWorkerForID(workerid)));
						if (((SimWorker) (activeWorkerPool.getWorkerForID(workerid))).getCorrectnessLevel() < StaticInformation.minimumCorrectnessLevel){
							addQualificationUpdate(workerid, 0);
						}else{
							addWorkerThread(((SimWorker) (activeWorkerPool
									.getWorkerForID(workerid))));
							// Add worker with qualification value
							// 1 means the worker may work because of right
							// qualification
							addQualificationUpdate(workerid, 1);
						}
					}

				} catch (Exception e) {
					basiclog.error("Error while reading file " + f);
				}
			}

			for (String s : hitIds)
				notPublishedHITs.add(s);

			// We are done with the initializing, now start the actual
			// simulation
			// running = true;
			// Thread t = new Thread(new SimMTurk());
			// t.start();

			isDoneWithInitializing = true;

			// running = true;
			// for (Thread wt : workerThreads)
			// wt.start();

		} else {
			basiclog.error("SIMINPUT directory is not a directory.");
		}
	}

	/**
	 * Simulates the publishing process of the HITs on a crowdsourcing platform.
	 */

	public void run() {

		running = true;
		for (Thread wt : workerThreads.values())
			wt.start();
		while (running) {
			try {
				Thread.sleep(StaticInformation.millisBetweenPublishingHITOnSimMTurk);
			} catch (InterruptedException e) {
				basiclog.error("SimMTurk was interrupted.");
			}

			int parallelHITs = publishedHITs.size()
					- (ResultOutputCollector.getCompletedHits().length + ResultOutputCollector
							.getEscalatedHits().length);

			if (parallelHITs < StaticInformation.maxParallelSimHITs
					&& !notPublishedHITs.isEmpty()) {
				String temp = notPublishedHITs.remove();
				basiclog.debug("Publishing new HIT " + temp);
				queuedHitIds.add(temp);
				publishedHITs.add(temp);
				basiclog.debug("parallelHITs: " + parallelHITs
						+ ", publishedHITs: " + publishedHITs.size()
						+ " not publishedHITs: " + notPublishedHITs.size());
			}
		}

		basiclog.debug("Exit SimMTurk");

	}

	/**
	 * @return the running
	 */
	public static boolean isRunning() {
		return running;
	}

	/**
	 * @param e
	 * @return
	 * @see java.util.Queue#add(java.lang.Object)
	 */
	public static boolean addHIT(String hitID) {
		basiclog.info("Adding hit number " + hitID);
		outputAnswers.put(hitID, new ConcurrentLinkedQueue<Assignment>());

		int parallelHITs = publishedHITs.size()
				- (ResultOutputCollector.getCompletedHits().length + ResultOutputCollector
						.getEscalatedHits().length);

		if (publishedHITs.contains(hitID)) {
			if (queuedHitIds.contains(hitID))
				basiclog.error("QueuedHitIdsObject enqueues " + hitID
						+ " again.");
			else
				queuedHitIds.add(hitID);
		} else if (parallelHITs < StaticInformation.maxParallelSimHITs
				&& notPublishedHITs.contains(hitID)) {
			basiclog.debug("Publishing hit number " + hitID);
			queuedHitIds.add(hitID);
			publishedHITs.add(hitID);
			notPublishedHITs.remove(hitID);
		}

		return true;
	}

	public static boolean extendHIT(String hitID) {
		try {

			// synchronized (queuedHitIds) {
			ArrayList<String> temp = new ArrayList<String>();
			while (queuedHitIds.size() > 0) {
				temp.add(queuedHitIds.remove());
			}
			queuedHitIds.add(hitID);
			for (String hit : temp) {
				queuedHitIds.add(hit);
			}
			// basiclog.debug("QueuedHitIds: " + queuedHitIds);
			if (!outputAnswers.containsKey(hitID)) {
				outputAnswers.put(hitID,
						new ConcurrentLinkedQueue<Assignment>());
			}
			basiclog.info("HIT " + hitID + " extended.");
			return true;
			// }
		} catch (Exception e1) {
			e1.printStackTrace();
			basiclog.warn("HIT " + hitID + " not extended.");
			return false;
		}
	}

	public static boolean extendHIT(Result r) {
		String hitID = r.getHitId();
		try {
			synchronized (prioritizedHITs) {
				synchronized (queuedHitIds) {
					prioritizedHITs.add(r);

					queuedHitIds.add(hitID);
					if (!outputAnswers.containsKey(hitID)) {
						outputAnswers.put(hitID,
								new ConcurrentLinkedQueue<Assignment>());
					}
					basiclog.info("HIT " + hitID + " extended.");
					return true;
				}
			}
			// }
		} catch (Exception e1) {
			e1.printStackTrace();
			basiclog.warn("HIT " + hitID + " not extended.");
			return false;
		}
	}

	/**
	 * @return
	 * @see java.util.Collection#isEmpty()
	 */
	public static boolean hasHITs() {
		return queuedHitIds.isEmpty();
	}

	/**
	 * @return
	 * @see java.util.Queue#remove()
	 */
	public static String getNextHIT() {
		String nextHIT = null;

		synchronized (queuedHitIds) {
			if (StaticInformation.scenario.getSimModeHIT() == SimMode.RANDOM) {
				String[] availableHITs = queuedHitIds
						.toArray(new String[queuedHitIds.size()]);

				int random = new Random().nextInt(availableHITs.length);
				nextHIT = availableHITs[random];
				queuedHitIds.remove(nextHIT);
			} else if (StaticInformation.scenario.getSimModeHIT() == SimMode.ORDER) {
				nextHIT = queuedHitIds.remove();
			}
		}

		basiclog.info("Retrieving hit number " + nextHIT + " (further "
				+ queuedHitIds.size() + " HITs exist)");
		return nextHIT;
	}

	public static String getNextHIT(SimWorker simWorker) {
		String nextHIT = null;
		boolean foundHIT = false;
		synchronized (prioritizedHITs) {
			synchronized (queuedHitIds) {

				Result debug = null;
				boolean noPrio = false;

				int tryNo = 0;
				while (!foundHIT && tryNo < 3) { // Got 3 trials to get a HIT

					if (StaticInformation.scenario.getSimModeHIT() == SimMode.RANDOM) {
						String[] availableHITs = queuedHitIds
								.toArray(new String[queuedHitIds.size()]);

						int random = new Random().nextInt(availableHITs.length);
						nextHIT = availableHITs[random];

						if (simWorker.hasAssignments(nextHIT)&& queuedHitIds.contains(nextHIT))
							foundHIT = true;
						else
							nextHIT = null;
					} else if (StaticInformation.scenario.getSimModeHIT() == SimMode.ORDER) {
						nextHIT = queuedHitIds.peek();
						foundHIT = true;
					} else if (StaticInformation.scenario.getSimModeHIT() == SimMode.PRIORITY) {
						Result[] orderedArray = prioritizedHITs
								.toArray(new Result[prioritizedHITs.size()]);
						Arrays.sort(orderedArray);

						if (!noPrio) {
							for (Result r : orderedArray) {
								nextHIT = r.getHitId();

								if (simWorker.hasAssignments(nextHIT)
										&& queuedHitIds.contains(nextHIT)) {
									foundHIT = true;
									prioritizedHITs.remove(r);
									debug = r;
									basiclog.debug("NextHIT is prioritized: "
											+ nextHIT);
									break;
								} else
									nextHIT = null;
							}
						}

						noPrio = (nextHIT == null);
						if (!foundHIT) {
							String[] availableHITs = queuedHitIds
									.toArray(new String[queuedHitIds.size()]);

							int random = new Random()
									.nextInt(availableHITs.length);
							nextHIT = availableHITs[random];

							if (simWorker.hasAssignments(nextHIT))
								foundHIT = true;
							else
								nextHIT = null;
						}
					}

					tryNo++;
				}
				if (queuedHitIds.contains(nextHIT)) {
					queuedHitIds.remove(nextHIT);
					basiclog.info("Retrieving hit number " + nextHIT
							+ " (further " + queuedHitIds.size()
							+ " HITs exist)");
					//return nextHIT;
				}/* else {
					basiclog.error("ERROR!!: QueuedHitIds: "
							+ queuedHitIds.toString() + " nextHIT: " + nextHIT);
					//prioritizedHITs.add(debug);
					//basiclog.error("ERROR!!: Rebuild old status! PrioritizedHITs: "
							+ prioritizedHITs.toString());
					return null;

				}*/
				return nextHIT;
			}
		}
	}

	/**
	 * @param e
	 * @return
	 * @see java.util.Queue#add(java.lang.Object)
	 */
	public static boolean addAnswer(String e, int hitId, String workerId,
			Calendar acceptTime) {
		counter++;
		basiclog.info("Adding answer " + e + " for HIT " + hitId + " ("
				+ counter + " answers in total).");
		return outputAnswers.get(hitId).add(
				createAssignment(e, hitId, workerId, acceptTime));
	}

	public static boolean addAnswer(String hitID, Assignment assignment) {
		counter++;
		boolean result = false;
		// synchronized(outputAnswers){
		result = outputAnswers.get(hitID).add(assignment);
		// }
		basiclog.info("Adding answer " + assignment.getAnswer() + " for HIT "
				+ hitID + " (" + counter + " answers in total).");
		return result;
	}

	public static void addWorkerThread(SimWorker simWorker) {
		if (!workerThreads.containsKey(simWorker.getWorkerId()))
			workerThreads.put(simWorker.getWorkerId(), new Thread(simWorker,
					simWorker.getWorkerId()));
	}

	/**
	 * @return
	 * @see java.util.Collection#isEmpty()
	 */
	public static boolean hasAnswers() {
		return outputAnswers.isEmpty();
	}

	/**
	 * @return
	 * @see java.util.Collection#remove()
	 */
	public static Assignment getNextAnswerForHit(String hitId) {
		return outputAnswers.get(hitId).remove();
	}

	/**
	 * @param key
	 * @param value
	 * @return
	 * @see java.util.Map#put(java.lang.Object, java.lang.Object)
	 */
	public static Integer addQualificationUpdate(String key, Integer value) {
		basiclog.debug("Qualification updated for worker " + key + " with "
				+ value);

		return qualificationValues.put(key, value);
	}

	/**
	 * @param key
	 * @return
	 * @see java.util.Map#remove(java.lang.Object)
	 */
	public static Integer getQualificationUpdate(Object key) {
		return qualificationValues.remove(key);
	}

	private static Assignment createAssignment(String answerValue, int hitid,
			String workerId, Calendar acceptTime) {
		Assignment a = new Assignment();
		a.setAnswer(buildXMLAnswer(answerValue));
		a.setHITId("" + hitid);
		a.setSubmitTime(Calendar.getInstance());
		a.setWorkerId(workerId);
		a.setAcceptTime(acceptTime);
		a.setAssignmentId("assignment_" + hitid + "_by_" + workerId);
		return a;
	}

	private static String buildXMLAnswer(String answerValue) {
		return "<QuestionFormAnswers xmlns=\"http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd\">"
				+ "<Answer>    <QuestionIdentifier>Dummy Question</QuestionIdentifier>    <FreeText>"
				+ answerValue + "</FreeText>  </Answer></QuestionFormAnswers>";
	}

	/**
	 * @return the hitIds
	 */
	public static String[] getHitIds() {
		return hitIds;
	}

	/**
	 * @return the hitIds
	 */
	public static String[] getPublishedHitIds() {
		return publishedHITs.toArray(new String[publishedHITs.size()]);
	}

	/**
	 * @return the queuedHitIds
	 */
	public static String[] getQueuedHitIds() {
		return queuedHitIds.toArray(new String[queuedHitIds.size()]);
	}

	public static boolean isDoneWithInitializing() {
		return isDoneWithInitializing;
	}

}
